package com.shujia.flink.source

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{ParallelSourceFunction, RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

object Demo5MysqlSource {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(2)

    //使用自定义source
    val mysqlDS: DataStream[Student] = env.addSource(new MysqlSource)

    mysqlDS
      .map(stu => (stu.clazz, 1))
      .keyBy(_._1)
      .sum(1)
      .print()


    env.execute()
  }

}

case class Student(id: String, name: String, age: Int, gender: String, clazz: String)


/**
  * SourceFunction； 单线程读取数据
  * ParallelSourceFunction： 并行读取数据，
  * RichSourceFunction 多了open 和cloase方法
  * RichSParallelSourceFunction
  */

class MysqlSource extends RichSourceFunction[Student] {


  var con: Connection = _

  /**
    * open 放在run方法之前执行
    *
    */
  override def open(parameters: Configuration): Unit = {
    //使jdbc读取mysql中的数据
    //1、加载驱动
    Class.forName("com.mysql.jdbc.Driver")

    //2、创建链接
    con = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata", "root", "123456")

  }

  /**
    * 在run 之后执行
    *
    */
  override def close(): Unit = {

    //关闭链接
    con.close()
  }

  /**
    * run在每一个并行的中只执行一次
    *
    */
  override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {


    //3、编写sql查询数据
    val stat: PreparedStatement = con.prepareStatement("select * from student")

    //4、执行查询
    val resultSet: ResultSet = stat.executeQuery()

    //5、解析数据
    while (resultSet.next()) {
      val id: String = resultSet.getString("id")
      val name: String = resultSet.getString("name")
      val age: Int = resultSet.getInt("age")
      val gender: String = resultSet.getString("gender")
      val clazz: String = resultSet.getString("clazz")

      val student = Student(id, name, age, gender, clazz)


      //将数据发送到下游
      ctx.collect(student)
    }


  }

  override def cancel(): Unit = {

  }
}